package org.tigr.htc.server.mw;

/*
This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation; either
version 2.1 of the License, or (at your option) any later version.

This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/

import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.Vector;

import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;
import org.apache.xmlrpc.WebServer;
import org.apache.xmlrpc.XmlRpcClient;
import org.tigr.htc.common.HTCConfig;
import org.tigr.htc.server.Scheduler;
import org.tigr.htc.server.condor.CondorUtils;

/**
 * The <b>MWServer</b> class runs a command of type "mw". mw commands are
 * array jobs where instead of a single command running on the grid the
 * command is a template for an array of identical tasks where
 * only the value of the arguments change.  The array of tasks can
 * be based on a file containing the arguments, a directory where each 
 * file in the directory is a argument, or a simple argument that is a count. 
 */

public class Server {
    static Logger log = Logger.getLogger(Server.class);
    
    static public Server mwserver;
    
    final static String MWURI = HTCConfig.getProperty("MWURI");
    
    /**
	 * The variable <code>peerList</code> keeps the list of peer servers that are active
	 */
    private List peerList = CondorUtils.peerList;
    
    //  client XML-RPC object to send procedure calls to server
    static private XmlRpcClient client;

    /**
     *  Private constructor so more than one server can't be created
     */
    private Server() { }


    /**
     * MWServer is a singleton and doesn't have a public constructor 
     *
     */
    public static Server getInstance() {
        if (mwserver == null) {
            mwserver = new Server();
        }

        return mwserver; 
    }

    @Override
    public Object clone() throws CloneNotSupportedException {
    	throw new CloneNotSupportedException();
    }

    public synchronized Hashtable getCommandHash(String cmd) {
        Thread.currentThread().setName("MW Command " + cmd);
        log.debug("In getCommandHash: " + cmd);

        try {
            long id = Long.parseLong(cmd);
            Runner runner = (Runner) Scheduler.getInstance().findRunner(id);

            if (runner != null) {
                return runner.getCommandHash();
            }
            else if (peerList.size() > 0) {
                // As there is no active runner and the command was not already complete
                // check with peers if they know anything about this job
                for (Iterator iter = peerList.iterator(); iter.hasNext();) {
                    String peer = (String) iter.next();
                    log.debug("Attempting to forward the getCommandHash request to peer " + peer);
                    client = (XmlRpcClient) CondorUtils.peerClients.get(peer);
                    try {
                        Vector params = new Vector();
                        // Add Command id
                        params.add(cmd); 
                        Boolean bool = (Boolean) client.execute("HTCServer.isMonitoring", params);
                        if (bool == Boolean.TRUE) {
                            // This peer is monitoring so forwarded stop request
                            if (log.isDebugEnabled()) {
                                log.debug("Forwarding getCommandHash request to peer.");
                            }
                            params = null;
                            params = new Vector();
                            // Add Command id
                            params.add(cmd); 
                            return (Hashtable) client.execute("MWServer.getCommandHash",params);
                        }
                    } catch (Exception e) {
                        log.warn("Error getting CommandHash from the peer for ." + cmd, e);
                        return null;
                    }
                } // End of for loop for peers
            } // end elseif 
            else {
                Hashtable result = new Hashtable();
                result.put("failure", "Couldn't find command " + cmd);
                return result;
            }
        } catch (Exception e) {
            log.debug("Failure in getCommandHash -" + e.getMessage() + "-", e);
            Hashtable result = new Hashtable();
            result.put("failure", e.getMessage());
            return result;
        }
        
        Hashtable result = new Hashtable();
        result.put("Failure", " couldn't find command " + cmd);
        return result;
    }
    

   /**
    * The method <code>getNextIndex</code> 
    *
    * @param cmd
    * @param slave
    * @param machine
    * @return
    */
    public synchronized int getNextIndex(String cmd, int slave, String machine) {
        Thread.currentThread().setName("MW Command " + cmd);
        log.debug("In getNextIndex:" + cmd);
        try {
            long id = Long.parseLong(cmd);
            Runner runner = (Runner) Scheduler.getInstance().findRunner(id);

            if (runner != null) {
                return runner.getNextIndex(slave, machine);
            }
            else if (peerList.size() > 0) {
                // As there is no active runner and the command was not already complete
                // check with peers if they know anything about this job
                for (Iterator iter = peerList.iterator(); iter.hasNext();) {
                    String peer = (String) iter.next();
                    log.debug("Attempting to forward getNextIndex request to peer " + peer);
                    client = (XmlRpcClient) CondorUtils.peerClients.get(peer);
                    try {
                        Vector params = new Vector();
                        // Add Command id
                        params.add(cmd); 
                        Boolean bool = (Boolean) client.execute("HTCServer.isMonitoring", params);
                        if (bool == Boolean.TRUE) {
                            // This peer is monitoring so forwarded stop request
                            if (log.isDebugEnabled()) {
                                log.debug("Forwarding getNextIndex request to peer.");
                            }
                            params = null;
                            params = new Vector();
                            // Add Command id
                            params.add(cmd); 
                            params.add(new Integer(slave));
                            params.add(machine);
                            
                            int nextIndex = ((Integer) client.execute("MWServer.getNextIndex", 
                                    										params)).intValue();
                            return nextIndex;
                        }
                    } catch (Exception e) {
                        log.warn("Error getting the next index from the peer.", e);
                        return -1;
                    }
                }
            }
            else {
              log.warn("Could not get the next index for command :" + cmd);
              return -1;   
            }
        } catch (Exception e) {
            log.debug("Failure in getNextIndex: ", e);
            return -1;
        }
        return -1;
    }
  
    public int getNextIndex(String p_cmd, Integer p_slave, String p_machine) {
        Thread.currentThread().setName("MW Command " + p_cmd);
        log.debug("In getNextIndex for " + p_cmd);
        try {
            long id = Long.parseLong(p_cmd);
            int slave = p_slave.intValue();
            Runner runner = (Runner) Scheduler.getInstance().findRunner(id);

            if (runner != null) {
                return runner.getNextIndex(slave, p_machine);
            } else {
              return -1;   
            }
        } catch (Exception e) {
            log.debug("Failure in getNextIndex.", e);
            return -1;
        }
    }

    
   /**
    * The method <code>setTaskFinished</code> 
    *
    * @param cmd
    * @param index
    * @param value
    * @param message
    * @return
    */
    public synchronized int setTaskFinished(String cmd, int index, int value, String message) {
        Thread.currentThread().setName("MW Command " + cmd);
        log.debug("In setTaskFinished " + cmd);
        try {
            long id = Long.parseLong(cmd);
            Runner runner = (Runner) Scheduler.getInstance().findRunner(id);
            
            // If runner is not null, delegate it to the runner.
            if (runner != null) {
                runner.setTaskFinished(index, value, message);
            } else if (peerList.size() > 0) {
                // As there is no active runner and the command was not already complete
                // check with peers if they know anything about this job
                for (Iterator iter = peerList.iterator(); iter.hasNext();) {
                    String peer = (String) iter.next();
                    log.debug("Attempting to setTaskFinished request to peer " + peer);
                    client = (XmlRpcClient) CondorUtils.peerClients.get(peer);
                    try {
                        Vector params = new Vector();
                        // Add Command id
                        params.add(cmd); 
                        Boolean bool = (Boolean) client.execute("HTCServer.isMonitoring", params);
                        if (bool == Boolean.TRUE) {
                            // This peer is monitoring so forwarded stop request
                            if (log.isDebugEnabled()) {
                                log.debug("Forwarding setTaskFinished request to peer.");
                            }
                            
                            // Create params for forwarding the request
                            params = null;
                            params = new Vector();
                            // Add Command id
                            params.add(cmd); 
                            params.add(new Integer(index));
                            params.add(new Integer(value));
                            params.add(message);
                            
                            int retVal = ((Integer)client.execute("MWServer.setTaskFinished", 
                                    										params)).intValue();
                            return retVal;
                        }
                    } catch (Exception e) {
                        log.warn("Error in setting the " + index + " task to be finished.", e);
                        return -1;
                    }
                } // End of looping through the peers
            } // end else
            return -1;
        } catch (Exception e) {
            log.debug(" failure in setTaskFinished " + e.getMessage(), e);
            return -1;
        }
    }
    
    
    /**
     * @param cmd
     * @param index
     * @param value
     * @param message
     * @return
     */
    public int setTaskFinished(String cmd, Integer index, Integer value, String message) {
        Thread.currentThread().setName("MW Command " + cmd);
        log.debug("In setTaskFinished.");

        int returnValue = 1;
        try {
            long id = Long.parseLong(cmd);
            Runner runner = (Runner) Scheduler.getInstance().findRunner(id);
            runner.setTaskFinished(index.intValue(), value.intValue(), message);
        } catch (Exception e) {
            log.debug(" failure in setReturnValue " + e.getMessage(), e);
            returnValue = -1;
        }
        return returnValue;
    }
    
    public int setTaskFailure(String cmd, Integer index, String message) {
        Thread.currentThread().setName("MW Command " + cmd);
        log.debug("In setTaskFailure: " + cmd);
        try {
            long id = Long.parseLong(cmd);
            Runner runner = (Runner) Scheduler.getInstance().findRunner(id);
            return runner.setTaskFailure(index.intValue(), message);
        } catch (Exception e) {
            log.debug("Failure in setTaskFailure: " + e.getMessage(), e);
            return -1;
        }
    }
    
    public synchronized int setTaskFailure(String cmd, int index, String message) {
        Thread.currentThread().setName("MW Command " + cmd);
        log.debug("In setTaskFailure: " + cmd);
        try {
            long id = Long.parseLong(cmd);
            Runner runner = (Runner) Scheduler.getInstance().findRunner(id);
            
            // If runner is not null, delegate the request to the runner object 
            // otherwise forward this request to one of the peers
            if (runner != null) {
                return runner.setTaskFailure(index, message);
            }
            else if (peerList.size() > 0) {
                // As there is no active runner and the command was not already complete
                // check with peers if they know anything about this job
                for (Iterator iter = peerList.iterator(); iter.hasNext();) {
                    String peer = (String) iter.next();
                    log.debug(" attempting to setTaskFinished request to peer " + peer);
                    client = (XmlRpcClient) CondorUtils.peerClients.get(peer);
                    try {
                        Vector params = new Vector();
                        // Add Command id
                        params.add(cmd); 
                        Boolean bool = (Boolean) client.execute("HTCServer.isMonitoring", params);
                        if (bool == Boolean.TRUE) {
                            // This peer is monitoring so forwarded stop request
                            if (log.isDebugEnabled()) {
                                log.debug("Forwarding setTaskFinished request to peer.");
                            }
                                
                            // Create params for forwarding the request
                            params = null;
                            params = new Vector();
                            // Add Command id
                            params.add(cmd); 
                            params.add(new Integer(index));
                            params.add(message);
                                
                            int retVal = ((Integer)client.execute("MWServer.setTaskFailure", 
                                    									params)).intValue();
                            return retVal;
                        }
                    } catch (Exception e) {
                        log.warn("Error in setting the " + index + " task to be finished.", e);
                        return -1;
                    }
                } // End of looping through the peers
            } // end else
        } catch (Exception e) {
            log.warn("Failure in setTaskFailure " + e.getMessage(), e);
            return -1;
        }
        return -1;
    }
    

    /**
     * For testing only, not for general use
     *
     */
    public static void main(String args[]) {
    	int PORT = 10888;
    	
    	try {
    		PropertyConfigurator.configure("conf/log4j_server.conf");
    		
    		log.debug("Create MW Server");
    		WebServer webserver = new WebServer(PORT);
    		
    		log.debug("Add handler");
    		webserver.addHandler("MWServer", mwserver);
    		
    		log.debug("Start");
    		webserver.start();
    		
    	} catch (Exception e) {
    		log.fatal("main -- " + e.getMessage(), e);
    	}
    }
}

